package groupOne.app.DWD.db;

import groupOne.app.BaseAppSql;
import groupOne.common.Constant;
import groupOne.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderDetail_huangyanhui extends BaseAppSql {
    public static void main(String[] args) {
        new DwdTradeOrderDetail_huangyanhui().init(
                4040,
                2,
                "DwdTradeOrderDetail_huangyanhui"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {

        //1. 读取预处理表
        tEnv.executeSql("create table dwd_trade_order_pre_process(" +
                " id string, " +
                " order_id string, " +
                " user_id string, " +
                " order_status string, " +
                " course_id string, " +
                " course_name string, " +
                " split_origin_amount string, " +
                " split_coupon_reduce string, " +
                " split_final_amount string, " +
                " create_time string, " +
                " date_id string, " +
                " update_date_id string, " +
                " update_time string, " +
                " origin_amount string, " +
                " coupon_reduce string, " +
                " final_amount string, " +
                " session_id string, " +
                " province_id string, " +
                " `type` string, " +
                " `old` map<string,string>, " +
                " od_ts bigint, " +
                " oi_ts bigint, " +
                " row_op_ts timestamp_ltz(3)" +
                " )" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_PRE_PROCESS,"DwdTradeOrderDetail_huangyanhui")
        );

        //2. 过滤出下单详情
        Table result = tEnv.sqlQuery("select " +
                " id, " +
                " order_id, " +
                " user_id, " +
                " order_status, " +
                " course_id, " +
                " course_name, " +
                " split_origin_amount, " +
                " split_coupon_reduce, " +
                " split_final_amount, " +
                " create_time, " +
                " date_id, " +
                " session_id, " +
                " province_id, " +
                " od_ts, " +
                " row_op_ts" +
                " from dwd_trade_order_pre_process " +
                " where `type`='insert'");
        //3.写出到kafka
        tEnv.executeSql("create table dwd_trade_order_detail( " +
                " id string, " +
                " order_id string, " +
                " user_id string, " +
                " order_status string, " +
                " course_id string, " +
                " course_name string, " +
                " split_origin_amount string, " +
                " split_coupon_reduce string, " +
                " split_final_amount string, " +
                " create_time string, " +
                " date_id string, " +
                " session_id string, " +
                " province_id string, " +
                " od_ts bigint, " +
                " row_op_ts timestamp_ltz(3) " +
                ")" +SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL)
        );
        result.executeInsert("dwd_trade_order_detail");

    }
}
